package com.tt.job.service;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.tt.job.JobInfo;
import com.tt.job.api.Job;
import com.tt.job.api.log.JobLoggerFactory;
import com.tt.job.common.util.SystemUtil;
import com.tt.job.container.ContainerSettings;
import com.tt.job.container.JobContainer;
import com.tt.job.register.JobChangeListener;
import com.tt.job.register.JobRegistryCenter;
import com.tt.job.rocketmq.MqConsumer;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 事件消费者服务，实现作为任务执行者时，消费事件执行具体的job
 */
@Service
public class JobEventConsumeService implements JobChangeListener {
    private final static Logger logger = LoggerFactory.getLogger(JobEventConsumeService.class);
    protected ConcurrentHashMap<String, JobInfo> registerJobs;
    @Autowired
    private ApplicationContext context;
    private ConcurrentHashMap<String, Job> localJobs = new ConcurrentHashMap<>();
    @Autowired
    private MqConsumer mqConsumer;
    @Autowired
    private JobRegistryCenter registryCenter;
    @Autowired
    private ContainerSettings settings;
    private boolean started = false;

    public void init() {

        //扫描所有本地包，得出本地可支持的所有任务信息
        Map<String, Job> jobBeans = context.getBeansOfType(Job.class);
        //扫描所有的容器中的job
        if (settings.isEnable()) {
            for (ContainerSettings.ContainerAppSettings appSettings : settings.getApps()) {
                JobContainer container = new JobContainer(appSettings.getName(), appSettings.getPath(), appSettings.getConfigClass());
                Map<String, Job> containerJobBeans = container.getJobMap();
                JobLoggerFactory.setLoggers(containerJobBeans, appSettings.getLevel());
                jobBeans.putAll(containerJobBeans);
            }
        }
        //将本地job转换为 jobClass->Job形式
        if (jobBeans != null && jobBeans.size() > 0) {
            for (Job job : jobBeans.values()) {
                if (checkJob(job))
                    localJobs.put(job.getClass().getName(), job);
            }
        }
        //从注册中心获取已注册的所有job信息 jobClass
        Map<String, JobInfo> registerJobs = registryCenter.getJobs();
        this.registerJobs = new ConcurrentHashMap<>(registerJobs);
        //对比差异，获取本地新增的job
        Collection<String> newJobs = CollectionUtils.subtract(localJobs.keySet(), registerJobs.keySet());
        //向注册中心新增本地新的job信息
        if (newJobs.size() > 0) {
            for (String newJob : newJobs) {
                registryCenter.addJob(newJob, localJobs.get(newJob));
            }
        }
        String localPid = String.format("%s@%s", SystemUtil.getLocalProcessId(), SystemUtil.getLocalAddress());
        //向注册中心添加job的client列表
        for (String job : localJobs.keySet()) {
            logger.debug("开始向注册中心注册job: {}", job);
            registryCenter.registryJobClient(job, localJobs.get(job).getType().toString(), localPid);
        }
        //TODO 更新其他job的信息

        logger.info("完成本地job注册");
    }

    /**
     * 校验job的实现是否符合需求
     *
     * @param job
     * @return
     */
    private boolean checkJob(Job job) {
        if (job == null) {
            logger.error("待添加的job为null");
            return false;
        }
        if (job.getType() == null || StringUtils.isEmpty(job.getName())) {
            logger.error("job的实现不符合要求，跳过。jobClass:{}", job.getClass().getName());
            return false;
        }
        return true;
    }

    @Override
    public void notifyJobAdded(String jobClass, Job job) {
        logger.debug("get job added notify. class:{}, job:{}", jobClass, job);
        if (!started) {
            logger.debug("job event consumer is not started!");
            return;
        }
        if (!localJobs.containsKey(jobClass)) {
            logger.debug("job event consumer local has no job:{}", jobClass);
            return;
        }
        //本地判断是否存在指定的job，如果存在，则启用job的工作，否则不做处理
        try {
            mqConsumer.subscribe(jobClass, localJobs.get(jobClass));
        } catch (MQClientException e) {
            logger.error("接收到job添加并订阅job:{} 的消息失败.", e);
        }
    }

    @Override
    public void notifyJobChanged(String jobClass, Job job) {
        logger.debug("get job change notify. class:{}, job:{}", jobClass, job);
        //job信息变化，无需处理
    }

    @Override
    public void notifyJobDeleted(String jobClass) {
        logger.debug("get job delete notify. class:{}", jobClass);
        if (!started) {
            return;
        }
        //判断本地是否存在指定的job，如果存在，则停止job的后续工作（终止任务执行），否则不做处理
        mqConsumer.unsubscribe(jobClass);
    }

    public synchronized void start() {
        if (started) {
            return;
        }
        started = true;
        //进行消息队列连接，并订阅本地能够处理的Job的消息
        for (Map.Entry<String, Job> jobEntry : localJobs.entrySet()) {
            try {
                mqConsumer.subscribe(jobEntry.getKey(), jobEntry.getValue());
            } catch (Exception e) {
                logger.error("订阅job:{} 的消息失败.", jobEntry.getKey(), e);
            }
        }
    }

    public synchronized void stop() {
        if (!started) {
            return;
        }
        //停止消息队列监听
        started = false;
        mqConsumer.unsubscribeAll();
    }
}
